package com.ruoyi.framework.task;

import com.ruoyi.project.ins.api.InsApi;
import com.ruoyi.project.ins.domain.*;
import com.ruoyi.project.ins.service.IInsAccountService;
import com.ruoyi.project.ins.service.IInsMessageService;
import com.ruoyi.project.ins.service.IInsPromotionService;
import com.ruoyi.project.ins.utils.HttpResult;
import com.ruoyi.project.ins.utils.TaskUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Component("insSendMessageTask")
@Slf4j
public class InsSendMessageTask {
    @Autowired
    private IInsAccountService insAccountService;

    @Autowired
    private IInsMessageService insMessageService;

    @Autowired
    private IInsPromotionService insPromotionService;

    // 注入线程池 Bean
    @Qualifier("asyncTaskExecutor")
    @Autowired
    private Executor asyncTaskExecutor;

    // ====== 批量更新缓存区 ======
    private final Queue<InsMessage> messageUpdateQueue = new ConcurrentLinkedQueue<>();
    private final Queue<InsPromotion> promotionUpdateQueue = new ConcurrentLinkedQueue<>();
    private final Queue<InsAccount> accountUpdateQueue = new ConcurrentLinkedQueue<>();

    public void run() {
        try {
            log.info("ins发消息定时任务开始执行");

            List<InsSendMessage> allMessages = insMessageService.selectNoSentMessage();
            if (CollectionUtils.isEmpty(allMessages)) {
                log.info("无需要推送的消息，本次操作跳过");
                return;
            }
            log.info("找到需要发送的消息数: {}", allMessages.size());

            List<InsPromotion> allPromotions = insPromotionService.selectNoPushedUser();
            if (CollectionUtils.isEmpty(allPromotions)) {
                log.info("无有效推送目标用户，任务退出");
                return;
            }

            long startTime = System.currentTimeMillis();
            // 按创建人分组，每组一个线程安全队列
            Map<String, Deque<InsPromotion>> promotionMap = allPromotions.stream().collect(
                Collectors.groupingBy(InsPromotion::getCreateBy, Collectors.toCollection(ConcurrentLinkedDeque::new))
            );

            Deque<InsPromotion> allUserDeque = new ConcurrentLinkedDeque<>(allPromotions);

            // 2️⃣ 批次并发执行消息
            List<CompletableFuture<Void>> futures = new ArrayList<>();
            int batchSize = TaskUtils.getBatchSize(allMessages.size());
            log.info("每组执行的任务数: {}", batchSize);
            int groupIndex = 1;
            AtomicInteger counter = new AtomicInteger(1);
            for (int i = 0; i < allMessages.size(); i += batchSize) {
                int end = Math.min(i + batchSize, allMessages.size());
                List<InsSendMessage> batchList = new ArrayList<>(allMessages.subList(i, end));
                log.info("即将启动发消息定时任务第{}组", groupIndex ++);
                // 每一批交给线程池的一个任务去串行处理
                CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                    for (InsSendMessage insMessage : batchList) {
                        log.info("发消息定时任务正在处理第{}条", counter.getAndIncrement());
                        processMessage(insMessage, promotionMap, allUserDeque);
                    }
                }, asyncTaskExecutor);
                // 保存 future，以便后面等待全部完成
                futures.add(future);
            }
            CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
            flushBatchUpdates();
            log.info("ins发消息定时任务执行结束, 耗时: {}s", (System.currentTimeMillis() - startTime) / 1000);
        } catch (Exception e) {
            log.info("ins发消息定时任务异常中断, e: {}", e.getMessage());
        }
    }

    private void processMessage(InsSendMessage insSendMessage, Map<String, Deque<InsPromotion>> promotionMap,
        Deque<InsPromotion> allUserDeque) {
        try {
            boolean sendSuccess;
            if (StringUtils.isNotEmpty(insSendMessage.getReceiveUserid())) {
                sendSuccess = requestSendMessage(insSendMessage);
            } else {
                String username = insSendMessage.getCreateBy();
                InsPromotion tmpInsPromotion;
                if (username.equalsIgnoreCase("admin")) {
                    tmpInsPromotion = allUserDeque.poll();
                    if (tmpInsPromotion == null) {
                        log.info("admin未获取到推广用户，跳过消息id: {}", insSendMessage.getId());
                        return;
                    }
                } else {
                    Deque<InsPromotion> deque = promotionMap.get(username);
                    if (deque == null || deque.isEmpty()) {
                        log.info("创建者 {} 没有可用推广用户，跳过消息id: {}", insSendMessage.getCreateBy(), insSendMessage.getId());
                        return;
                    }
                    tmpInsPromotion = deque.poll();
                    if (tmpInsPromotion == null) {
                        log.info("创建者 {} 未获取到推广用户，跳过消息id: {}", insSendMessage.getCreateBy(), insSendMessage.getId());
                        return;
                    }
                }
                insSendMessage.setReceiveUserid(tmpInsPromotion.getUserId());
                insSendMessage.setReceiveUsername(tmpInsPromotion.getUsername());
                sendSuccess = requestSendMessage(insSendMessage);
            }

            if (sendSuccess) {
                insSendMessage.setStatus("1");
                messageUpdateQueue.add(InsMessage.fromInsSendMessage(insSendMessage));

                InsPromotion promotion = new InsPromotion();
                promotion.setUserId(insSendMessage.getReceiveUserid());
                promotion.setStatus("1");
                promotionUpdateQueue.add(promotion);

                InsAccount acc = new InsAccount();
                acc.setUserId(insSendMessage.getSendUserid());
                accountUpdateQueue.add(acc);
            } else {
                int failTimes = insSendMessage.getFailTimesOrDefault() + 1;
                insSendMessage.setFailTimes(failTimes);
                insSendMessage.setStatus(failTimes >= 2 ? "3" : "2");
                messageUpdateQueue.add(InsMessage.fromInsSendMessage(insSendMessage));

                InsPromotion promotion = new InsPromotion();
                promotion.setUserId(insSendMessage.getReceiveUserid());
                promotion.setStatus("2");
                promotionUpdateQueue.add(promotion);
            }

        } catch (Exception e) {
            log.error("处理消息异常，消息id: {}, 异常: {}", insSendMessage.getId(), e.getMessage(), e);
        }
    }

    private boolean requestSendMessage(InsSendMessage insSendMessage) {
        try {
            InstagramAccount account = InstagramAccount.builder()
                .userId(insSendMessage.getSendUserid()).bearerToken(insSendMessage.getAuth())
                .rur(insSendMessage.getRur()).wwwClaim(insSendMessage.getClaim()).mid(insSendMessage.getMid())
                .deviceId(insSendMessage.getIgDid()).uuidStr(insSendMessage.getIgDid())
                .build();

            InsWebProxy insWebProxy = new InsWebProxy();
            insWebProxy.setProxyServer(insSendMessage.getProxyServer());
            insWebProxy.setProxyPort(insSendMessage.getProxyPort());
            insWebProxy.setProxyCustomer(insSendMessage.getProxyCustomer());
            insWebProxy.setProxyPassword(insSendMessage.getProxyPassword());
            insWebProxy.setCountryCode(insSendMessage.getCountryCode());
            String messageType = insSendMessage.getMessageType();
            HttpResult httpResult;
            if (messageType.equals("1")) {
                // 发送私信
                httpResult = InsApi.smartSend(account, insSendMessage.getMessageContent(), null,
                    insSendMessage.getReceiveUserid(), null, insWebProxy);
            } else if (messageType.equals("2")) {
                // 发送名片+私信
                httpResult = InsApi.smartSend(account, insSendMessage.getMessageContent(), null,
                    insSendMessage.getReceiveUserid(), insSendMessage.getShareUserid(), insWebProxy);
            } else {
                // 发送名片
                httpResult = InsApi.smartSend(account, null, null, insSendMessage.getReceiveUserid(),
                    insSendMessage.getShareUserid(), insWebProxy);
            }
            log.info("发送消息返回结果: {}", httpResult);
            if (httpResult.getStatusCode() != 200) {
                log.info("发送消息失败");
                return false;
            }
            log.info("发送消息成功");
            return true;
        } catch (Exception e) {
            log.error("发送消息错误: {}", e.getMessage());
            return false;
        }
    }

    // ====== 批量刷新方法 ======
    private synchronized void flushBatchUpdates() {
        try {
            if (!messageUpdateQueue.isEmpty()) {
                List<InsMessage> messageList = new ArrayList<>();
                while (!messageUpdateQueue.isEmpty()) {
                    messageList.add(messageUpdateQueue.poll());
                }
                insMessageService.batchUpdateMessages(messageList);
                log.info("批量更新消息表 {} 条", messageList.size());
            }

            if (!promotionUpdateQueue.isEmpty()) {
                List<InsPromotion> promotionList = new ArrayList<>();
                while (!promotionUpdateQueue.isEmpty()) {
                    promotionList.add(promotionUpdateQueue.poll());
                }
                insPromotionService.batchUpdatePromotions(promotionList);
                log.info("批量更新推广用户表 {} 条", promotionList.size());
            }

            if (!accountUpdateQueue.isEmpty()) {
                List<InsAccount> accountList = new ArrayList<>();
                while (!accountUpdateQueue.isEmpty()) {
                    accountList.add(accountUpdateQueue.poll());
                }
                insAccountService.batchUpdateSuccessTimes(accountList);
                log.info("批量更新账号表 {} 条", accountList.size());
            }

        } catch (Exception e) {
            log.error("批量刷新更新异常: {}", e.getMessage(), e);
        }
    }
}
